package com.atguigu.flink.chapter04;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * TODO Operator Chain
 *
 * @author cjp
 * @version 1.0
 * @date 2021/8/6 11:16
 */
public class Flink03_WC_OperatorChain {
    public static void main(String[] args) throws Exception {
        // 0.获取 执行环境
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);
        env.disableOperatorChaining();
        // 1.读取数据
        DataStreamSource<String> socketDS = env.socketTextStream("localhost", 9999);
        // 2.处理数据 3.输出(打印)
        socketDS
                .flatMap(new FlatMapFunction<String, String>() {
                    @Override
                    public void flatMap(String value, Collector<String> out) throws Exception {

                        String[] words = value.split(" ");
                        for (String word : words) {
                            out.collect(word);
                        }
                    }
                })
//                .disableChaining()
                .map(new MapFunction<String, Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(String value) throws Exception {
                        return Tuple2.of(value, 1);
                    }
                })
                .keyBy(0)
                .sum(1)
                .print();

        // 4.启动
        env.execute("");

    }
}
/*

    Task：就是subtask的一次执行
    subTask： 算子的一个并行实例

    Operator Chain： 并行度相同，并且满足 One-to One的关系

    A     B     ，并行度都是3，满足 one to one
    ⭕    口
    ⭕    口
    ⭕    口

    ====》 将他们的 子任务 组合起来（串成链条）

    （⭕  -》口）   ==》 串起来之后，还是叫做 subtask
    （⭕  -》口）   ==》 串起来之后，还是叫做 subtask
    （⭕  -》口）   ==》 串起来之后，还是叫做 subtask

    算子.disableChaining :当前算子不参与 链化，包括前面和后面
    env.disableOperatorChaining: 全局禁用 链化

    企业怎么用？
        1、当某个算子逻辑比较复杂，不希望跟其他的串起来，可以断链，结合slot共享组单独跑
        2、反压时，排查问题，先断链
 */